package com.neusoft.bd.myflink.step05;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;
import java.util.Objects;

import com.neusoft.bd.myflink.entity.WaterSensor;

public class StreamFromArray {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        List<WaterSensor> data = Arrays.asList(
                new WaterSensor("sensor_1", 1, 1),
                new WaterSensor("sensor_1", 2, 2),
                new WaterSensor("sensor_2", 2, 2),
                new WaterSensor("sensor_3", 3, 3));
        DataStreamSource<WaterSensor> ds = env.fromCollection(data);
        env.setParallelism(2);

        SingleOutputStreamOperator<String> map = ds.map(new RichMapFunction<WaterSensor, String>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                RuntimeContext runtimeContext = getRuntimeContext();
                System.out.println("NumberOfParallelSubtasks->" + runtimeContext.getNumberOfParallelSubtasks()
                        + runtimeContext.getIndexOfThisSubtask() + " 的任务的生命周期开始");

            }

            @Override
            public String map(WaterSensor ws) throws Exception {
                return ws.toString();
            }

            @Override
            public void close() throws Exception {
                super.close();
            }
        });

        ds.print();
        env.execute();
    }
}
